package mqx

import (
	"fmt"
	"strings"

	"github.com/nsqio/go-nsq"
	"zhonghui/console/warehouse/cfgx"
	"zhonghui/console/warehouse/envx"
	"zhonghui/console/warehouse/logx"
)

func init() {
	cfgx.Fetch(cfgx.CfgItem{
		Id:    "mq",
		Group: "common",
	})

	// setupMns()
}

// 实现nsq日志
type nsqLogger struct{}

func (nl *nsqLogger) Output(callDepth int, s string) error {
	if strings.Contains(s, "ERR") {
		logx.Zap().Errorw("nsq日志",
			"callDepth", callDepth,
			"info", s,
		)
	} else {
		logx.Zap().Infow("nsq日志",
			"callDepth", callDepth,
			"info", s,
		)
	}
	return nil
}

func nsqConfig() *nsq.Config {
	c := nsq.NewConfig()
	// 设置默认最大接收数
	c.MaxInFlight = 3

	return c
}

// NewNsqProducer 初始化nsq producer
func NewNsqProducer() (*nsq.Producer, error) {
	cfg := cfgx.Cfg().GetStringMapString("mq.nsq.producer")

	producer, err := nsq.NewProducer(cfg["host"]+":"+cfg["port"], nsqConfig())
	if err != nil {
		return nil, err
	}

	// 设置自带的log
	producer.SetLogger(&nsqLogger{}, nsq.LogLevelInfo)

	return producer, nil
}

// StartNsqConsumer 接收nsq的消息
// func (h *MyMessageHandler) HandleMessage(m *nsq.Message) error
func StartNsqConsumer(handler nsq.Handler, topicName, channelName string, concurrency int) {
	defer func() {
		if r := recover(); r != nil {
			logx.Zap().Errorw("启动消费者异常",
				"err", fmt.Sprintf("%s", r),
			)
		}
	}()

	cfg := cfgx.Cfg().GetStringMapString("mq.nsq.consumer")

	nsqCfg := nsqConfig()

	// 重新设置最大接收数
	if concurrency > 0 {
		nsqCfg.MaxInFlight = concurrency
	}

	// 初始化消费者
	consumer, err := nsq.NewConsumer(topicName, channelName, nsqCfg)
	if err != nil {
		panic(err)
	}

	// 设置自带的log
	consumer.SetLogger(&nsqLogger{}, nsq.LogLevelInfo)
	// 设置处理函数
	consumer.AddConcurrentHandlers(handler, concurrency)

	// 订阅
	// 都走lookup
	if !envx.InDev() {
		// 生产环境使用链接 nsqlookup
		if err := consumer.ConnectToNSQLookupds([]string{cfg["host"] + ":" + cfg["port"]}); err != nil {
			panic(err)
		}
	} else {
		// 非生产环境使用链接 nsqd
		if err := consumer.ConnectToNSQDs([]string{cfg["host"] + ":" + cfg["port"]}); err != nil {
			panic(err)
		}
	}

	// 阻塞
	<-consumer.StopChan
}
